package org.zoomdev.zoom.rpc.client;

import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.zoomdev.zoom.common.exceptions.ZoomException;
import org.zoomdev.zoom.rpc.MethodId;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

public class ClientConnection {

    private static final Log log = LogFactory.getLog(RpcClient.class);
    private final List<SimpleHandler> handlers = new ArrayList<>();
    private final Map<Integer, Completer> completerMap = new ConcurrentHashMap<>();
    private final AtomicInteger counter = new AtomicInteger();
    private volatile boolean connected;

    private volatile boolean connecting;
    private final RpcClient client;
    private Channel channel;

    private final RpcService service;

    private final Map<String, Integer> ids = new ConcurrentHashMap<>();
    private Handler handler;


    public RpcService getService() {
        return service;
    }

    private final RpcClientListener listener;


    private boolean ready;

    public ClientConnection(RpcService service, RpcClient client, RpcClientListener listener) {
        this.client = client;
        this.service = service;
        this.listener = listener;
    }


    public void connect() {
        if (connected || connecting) {
            return;
        }
        connecting = true;
        if (client.listener != null) {
            client.listener.onRpcConnecting(this.service);
        }
        //log.info("rpc正在连接服务器" + this.service.toAddress());
        client.bootstrap.connect(this.service.toAddress()).addListener(new ChannelFutureListener() {
                    @Override
                    public void operationComplete(ChannelFuture channelFuture) throws Exception {
                        onConnect(channelFuture);
                    }
                })
                .channel()
                .closeFuture()
                .addListener(new ChannelFutureListener() {
                    @Override
                    public void operationComplete(ChannelFuture channelFuture) throws Exception {
                        onClose(channelFuture);
                    }
                });
    }

    protected void onClose(ChannelFuture future) {
        //System.out.println(future.cause());
        reconnect();
    }


    protected void onConnect(ChannelFuture future) {
        if (future.isSuccess()) {
            channel = future.channel();
            ClientHandler handler = channel.pipeline().get(ClientHandler.class);
            ClientIdleHandler idleHandler = channel.pipeline().get(ClientIdleHandler.class);
            idleHandler.setConnection(ClientConnection.this);
            handler.setConnection(ClientConnection.this);
            connecting = false;
            connected = true;
            //log.info("rpc连接服务成功");
            if (listener != null) {
                listener.onRpcConnected();
            }
        } else {
            if (listener != null) {
                listener.onRpcConnectionException(future.cause());
            }
        }
    }

    protected void addToQueue(Completer completer) {
        synchronized (paddingQueue) {
            paddingQueue.add(completer);
        }
    }


    protected void executeQueue() {
        if (channel == null) {
            throw new ZoomException("不可能的异常,channel==null");
        }

        synchronized (paddingQueue) {
            for (Completer completer : paddingQueue) {
                completer.func = completer.handler.func;
                if (!completer.isDone()) {
                    channel.writeAndFlush(completer);
                }
            }
            paddingQueue.clear();
            channel.flush();
        }
    }

    protected void clearQueue() {
        synchronized (paddingQueue) {
            IOException e = new IOException(connecting ? "rpc服务连接失败:" + service.toAddress() : "rpc连接断开:" + service.toAddress());
            for (Completer completer : completerMap.values()) {
                completer.setException(e);
            }

            for (Completer completer : paddingQueue) {
                completer.setException(e);
            }

            paddingQueue.clear();
            completerMap.clear();
        }

    }

    private final List<Completer> paddingQueue = new ArrayList<>();


    protected Completer create(short func, byte[] bytes) {
        int id = counter.incrementAndGet();
        Completer completer = new Completer(client.timeoutMs);
        completer.id = id;
        completer.func = func;
        completer.bytes = bytes;
        completerMap.put(id, completer);
        return completer;
    }

    public Completer call(SimpleHandler handler, short func, byte[] bytes) {
        Completer completer = create(func, bytes);
        completer.handler = handler;
        //func在哪个服务上面?
        if (client.boss != null) {
            client.boss.execute(() -> {
                if (!ready || completer.id == 0) {
                    addToQueue(completer);
                } else {
                    channel.writeAndFlush(completer);
                }
            });
        }

        return completer;
    }


    public Completer getCompleter(int id) {
        return completerMap.remove(id);
    }


    public void reconnect() {
        clearQueue();
        connecting = false;
        connected = false;
        if (client.boss != null) {
            client.boss.schedule(() -> {
                connect();
            }, 3000, TimeUnit.MILLISECONDS);
        }

    }


    public void setHeartbeatHandler(Handler handler) {
        this.handler = handler;
    }

    public void heartBeat() {
        int n = ThreadLocalRandom.current().nextInt();
        try {
            handler.handle(new Object[]{n}).then(o -> {
                int m = (Integer) o;
                if (m != n && channel != null && channel.isOpen()) {
                    channel.close();
                }
                return o;
            });
        } catch (Throwable ex) {
            channel.close();
        }


    }


    protected Map<String, Short> idMap = new ConcurrentHashMap<>();

    public void setMethodIds(List<MethodId> list) {
        for (MethodId id : list) {
            idMap.put(id.method, id.id);
        }

        for (SimpleHandler handler : handlers) {
            Short id = idMap.get(handler.method.toGenericString());
            if (id == null) {
                throw new ZoomException("rpc服务端不存在方法:" + handler.method.toGenericString());
            }
            handler.func = id;
        }

        ready = true;
        executeQueue();

    }

    public void addHandler(SimpleHandler handler) {
        handlers.add(handler);
    }
}
